use tokio::{io::{AsyncReadExt}, net::{TcpListener, TcpStream}};
use crate::mq::broker::{self};

const PACKAGE_END_FALG:u8 = 0x01;

#[derive(Clone)]
pub struct TcpServer{
    addr:String,
    raw_mode:bool,
}

impl TcpServer{
    pub fn new(addr:String,raw_mode: bool)->Self{
        TcpServer{addr,raw_mode}
    }
    async fn loop_accept(&self,listener: TcpListener, raw_mode: bool){
        loop {
            // 接受客户端连接
            let client = listener.accept().await;
            match client {
                Ok(s) => {
                    self.handle_client(s.0, raw_mode).await;
                }
                Err(err) => {
                    println!("Error:{:?}", err);
                }
            }
        }
    }

    pub async fn start(&self){
        let listener = TcpListener::bind(self.addr.clone()).await;
        match listener {
            Ok(t) => {
                let run = self.raw_mode;
                let self1 = self.clone();
                tokio::spawn(async move {
                    self1.loop_accept(t, run).await;
                });
            }
            Err(err) => {
                println!("Error:{:?}", err);
            }
        }
    }
    async fn handle_client(&self,stream: TcpStream, raw_mode: bool){
        let self1 = self.clone();
        tokio::spawn(async move {
            if raw_mode {
                self1.process_raw_client(stream).await;
            } else {
                self1.process_package_client(stream).await;
            }
        });
    }

    async fn process_raw_client(&self,stream: TcpStream){
        let addr = stream.peer_addr().unwrap().to_string();
        let sc =
        {
            let st = broker::get_broker();
            let mut s = st.lock().await;
            let aa = s.add_client(stream);
            aa.unwrap().clone()
        };
        
        let mut buffer = vec![0; 1024];
        loop {
            match sc.lock().await.read(&mut buffer).await {
                Ok(0) => {
                    println!("Client disconnected");
                    break;
                }
                Ok(n) => {
                    let data = buffer[..n].to_vec();
                    {
                        let st = broker::get_broker();
                        let mut s = st.lock().await;
                        s.process_client_data(&addr, data).await;
                    }
                }
                Err(err) => {
                    println!("Error reading from client: {:?}", err);
                    break;
                }
            }
        }
        {
            let st = broker::get_broker();
            let mut s = st.lock().await;
            s.remove_client(&addr).await;
        }
    }

    async fn process_package_client(&self,stream: TcpStream){
        let addr = stream.peer_addr().unwrap().to_string();
        let sc =
        {
            let st = broker::get_broker();
            let mut s = st.lock().await;
            let cc = s.add_client(stream);
            cc.unwrap().clone()
        };
        loop {
            let res = sc.lock().await.read_u32().await;
            match res {
                Ok(len) => {
                    let mut data = vec![0u8; len as usize];
                    let cc = sc.lock().await.read_exact(&mut data).await;
                    match cc {
                        Ok(0) => {
                            println!("Client disconnected");
                            break;
                        }
                        Ok(ll) => {
                            if data[ll - 1] == PACKAGE_END_FALG {
                                let res = Vec::from(&data[0..ll - 1]);
                                {
                                    let st = broker::get_broker();
                                    let mut s = st.lock().await;
                                    s.process_client_data(&addr, res).await;
                                }
                            } else {
                                println!("Package End Flag Error");
                                break;
                            }
                        }
                        Err(err) => {
                            println!("DisConnect,Error:{:?}", err);
                            break;
                        }
                    }
                }
                Err(err) => {
                    println!("DisConnect,Error:{:?}", err);
                    break;
                }
            };
        }
        {
            let st = broker::get_broker();
            let mut s = st.lock().await;
            s.remove_client(&addr).await;
        }
    }
}